package org.timewindow

import org.FlinkStreamApp
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.diysource.SensorSource

/**
 * description ：需求：计算窗口的温度平均值
 * 使用增量聚合的方式计算平均值
 * 教程 6-11
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/23 10:15
 * modified By ：
 * version:    : 1.0
 */
object AggregateFunctionExample extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    // 获取源数据
    val source = environment.addSource(new SensorSource)

    // 计算
    val minTempStream = source
      .map(l => (l.id, l.temperature))
      .keyBy(_._1)
      .timeWindow(Time.seconds(15))
      .aggregate(new AvgTempFunction)

    // 输出结果
    minTempStream.print
  }

  // ACC 计数器类型 (String, Double, Int) (id, 温度总和， 出现次数)
  class AvgTempFunction extends AggregateFunction[(String, Double), (String, Double, Int), (String, Double)] {
    override def createAccumulator() = ("", 0.0, 0)

    override def add(value: (String, Double), accumulator: (String, Double, Int)) = (value._1, accumulator._2 + value._2, accumulator._3 + 1)

    override def getResult(accumulator: (String, Double, Int)) = (accumulator._1, accumulator._2 / accumulator._3)

    override def merge(a: (String, Double, Int), b: (String, Double, Int)) = (a._1, a._2 + b._2, a._3 + b._3)
  }

}
